package bjut.zrpc.core.optimization.zookeeper;


import bjut.zrpc.core.optimization.enums.ZRpcConfigEnum;
import bjut.zrpc.core.optimization.event.*;
import bjut.zrpc.core.optimization.extension.ExtensionLoader;
import bjut.zrpc.core.optimization.factory.SingletonFactory;
import bjut.zrpc.core.optimization.interfaces.LoadBalance;
import bjut.zrpc.core.optimization.interfaces.ServiceDiscover;
import bjut.zrpc.core.optimization.protect.ServicesRecord;
import bjut.zrpc.core.optimization.message.ZRpcRequestMessage;
import bjut.zrpc.core.optimization.utils.CuratorUtil;
import bjut.zrpc.core.optimization.utils.PropertiesReadUtil;
import org.apache.curator.framework.CuratorFramework;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;

/**
 * @Author: zhuzhiming
 * @Date: 2022/6/24 18:48
 */
public class ZookeeperDiscover implements ServiceDiscover, ZRpcEventPublisher {

    private final ZookeeperRegistry zookeeperRegistry;
    private final LoadBalance loadBalance;
    private SimpleZRpcEventMulticaster zRpcEventMulticaster;


    public ZookeeperDiscover() {
        zookeeperRegistry = SingletonFactory.getInstance(ZookeeperRegistry.class);
        String loadBalance = PropertiesReadUtil.getConfig(ZRpcConfigEnum.ZRPC_LOADBALANCE.getPropertyValue()[0]);
        this.loadBalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(loadBalance);
        zRpcEventMulticaster = SingletonFactory.getInstance(SimpleZRpcEventMulticaster.class);
    }


    @Override
    public InetSocketAddress lookupService(ZRpcRequestMessage zRpcRequestMessage) {

        InetSocketAddress inetSocketAddress;
        String[] serviceAddress;
        String directAddress = PropertiesReadUtil.getConfig(ZRpcConfigEnum.ZRPC_DIRECT_SERVER_ADDRESS.getPropertyValue()[0]);

        if (!directAddress.equals("true")) {
            serviceAddress = directAddress.split(":");
            inetSocketAddress = new InetSocketAddress(serviceAddress[0], Integer.parseInt(serviceAddress[1]));
            return inetSocketAddress;
        }

        String serviceInfo = zRpcRequestMessage.getServiceInfo();

        CuratorFramework zkClient = CuratorUtil.getZkClient();


        List<String> serviceList = ServicesRecord.SERVICES_MAP.get(serviceInfo);
        if (serviceList == null || serviceList.size() == 0) {
            serviceList = CuratorUtil.getChildrenNodes(zkClient, serviceInfo);
            ServicesRecord.SERVICES_MAP.put(serviceInfo, new ArrayList<>(serviceList));
        }

        String service;

        publishEvent(new PullServiceBeforeEvent(this));

        try {

            service = loadBalance.selectService(serviceList, zRpcRequestMessage);
            serviceAddress = service.split(":");

            publishEvent(new PullServiceAfterEvent(this));

        } catch (Exception e) {
            throw new RuntimeException("no available service");
        }

        inetSocketAddress = new InetSocketAddress(serviceAddress[0], Integer.parseInt(serviceAddress[1]));

        return inetSocketAddress;

    }


    @Override
    public void publishEvent(ZRpcEvent event) {
        zRpcEventMulticaster.multicastEvent(event);
    }
}
